package com.zzh.partnersys.es.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.DeleteOperation;
import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
import cn.hutool.core.collection.CollUtil;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.zzh.partnersys.entity.TeamDO;
import com.zzh.partnersys.entity.UserDO;
import com.zzh.partnersys.es.entity.TeamDocument;
import com.zzh.partnersys.es.entity.UserDocument;
import com.zzh.partnersys.es.service.ElasticsearchSyncService;
import com.zzh.partnersys.mapper.TeamMapper;
import com.zzh.partnersys.mapper.UserMapper;
import com.zzh.partnersys.mapper.UserTeamMapper;
import com.zzh.partnersys.service.UserTeamService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.lang.reflect.Type;
import java.util.*;
import java.util.stream.Collectors;

/**
 * ES 数据同步服务实现
 *
 * @author zzh
 */
@Slf4j
@Service
public class ElasticsearchSyncServiceImpl implements ElasticsearchSyncService {

    private static final String USER_INDEX = "user_index";
    private static final String TEAM_INDEX = "team_index";

    @Resource
    private ElasticsearchClient elasticsearchClient;

    @Resource
    private UserMapper userMapper;

    @Resource
    private TeamMapper teamMapper;

    @Resource
    private UserTeamMapper userTeamMapper;

    @Resource
    private UserTeamService userTeamService;

    private final Gson gson = new Gson();
    private static final Type LIST_STRING_TYPE = new TypeToken<List<String>>() {}.getType();

    @Override
    @Async
    public void syncUserToEs(UserDO userDO) {
        if (userDO == null || userDO.getId() == null) {
            log.warn("同步用户到 ES 失败：用户数据为空");
            return;
        }

        try {
            // 转换为 ES 文档
            UserDocument userDocument = convertToUserDocument(userDO);

            // 使用 IndexRequest 进行 upsert 操作（存在则更新，不存在则创建）
            IndexRequest<UserDocument> indexRequest = IndexRequest.of(i -> i
                    .index(USER_INDEX)
                    .id(String.valueOf(userDO.getId()))
                    .document(userDocument)
            );

            elasticsearchClient.index(indexRequest);
            log.info("同步用户到 ES 成功，userId: {}", userDO.getId());
        } catch (IOException e) {
            log.error("同步用户到 ES 失败，userId: {}", userDO.getId(), e);
        }
    }

    @Override
    @Async
    public void syncTeamToEs(TeamDO teamDO) {
        if (teamDO == null || teamDO.getId() == null) {
            log.warn("同步队伍到 ES 失败：队伍数据为空");
            return;
        }

        try {
            // 转换为 ES 文档（包含重组后的 tags）
            TeamDocument teamDocument = convertToTeamDocument(teamDO);

            // 使用 IndexRequest 进行 upsert 操作
            IndexRequest<TeamDocument> indexRequest = IndexRequest.of(i -> i
                    .index(TEAM_INDEX)
                    .id(String.valueOf(teamDO.getId()))
                    .document(teamDocument)
            );

            elasticsearchClient.index(indexRequest);
            log.info("同步队伍到 ES 成功，teamId: {}", teamDO.getId());
        } catch (IOException e) {
            log.error("同步队伍到 ES 失败，teamId: {}", teamDO.getId(), e);
        }
    }

    @Override
    @Async
    public void updateTeamTags(Long teamId) {
        if (teamId == null || teamId <= 0) {
            log.warn("更新队伍 tags 失败：teamId 无效");
            return;
        }

        try {
            // 查询队伍信息
            TeamDO teamDO = teamMapper.selectById(teamId);
            if (teamDO == null) {
                log.warn("更新队伍 tags 失败：队伍不存在，teamId: {}", teamId);
                return;
            }

            // 重新同步队伍（会重新计算 tags）
            syncTeamToEs(teamDO);
            log.info("更新队伍 tags 成功，teamId: {}", teamId);
        } catch (Exception e) {
            log.error("更新队伍 tags 失败，teamId: {}", teamId, e);
        }
    }

    @Override
    @Async
    public void deleteUserFromEs(Long userId) {
        if (userId == null || userId <= 0) {
            log.warn("删除 ES 用户失败：userId 无效");
            return;
        }

        try {
            elasticsearchClient.delete(d -> d
                    .index(USER_INDEX)
                    .id(String.valueOf(userId))
            );
            log.info("删除 ES 用户成功，userId: {}", userId);
        } catch (IOException e) {
            log.error("删除 ES 用户失败，userId: {}", userId, e);
        }
    }

    @Override
    @Async
    public void deleteTeamFromEs(Long teamId) {
        if (teamId == null || teamId <= 0) {
            log.warn("删除 ES 队伍失败：teamId 无效");
            return;
        }

        try {
            elasticsearchClient.delete(d -> d
                    .index(TEAM_INDEX)
                    .id(String.valueOf(teamId))
            );
            log.info("删除 ES 队伍成功，teamId: {}", teamId);
        } catch (IOException e) {
            log.error("删除 ES 队伍失败，teamId: {}", teamId, e);
        }
    }

    @Override
    @Async
    public void batchSyncUsersToEs(List<UserDO> userDOList) {
        if (CollUtil.isEmpty(userDOList)) {
            log.warn("批量同步用户到 ES 失败：用户列表为空");
            return;
        }

        try {
            BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
            
            for (UserDO userDO : userDOList) {
                if (userDO == null || userDO.getId() == null) {
                    continue;
                }
                
                UserDocument userDocument = convertToUserDocument(userDO);
                IndexOperation<UserDocument> indexOp = IndexOperation.of(i -> i
                        .index(USER_INDEX)
                        .id(String.valueOf(userDO.getId()))
                        .document(userDocument)
                );
                bulkBuilder.operations(BulkOperation.of(o -> o.index(indexOp)));
            }

            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkBuilder.build());
            
            if (bulkResponse.errors()) {
                bulkResponse.items().forEach(item -> {
                    if (item.error() != null) {
                        log.error("批量同步用户到 ES 失败：文档 ID={}，错误原因={}", 
                                item.id(), item.error().reason());
                    }
                });
            } else {
                log.info("批量同步用户到 ES 成功，数量={}", userDOList.size());
            }
        } catch (IOException e) {
            log.error("批量同步用户到 ES 失败", e);
        }
    }

    @Override
    @Async
    public void batchSyncTeamsToEs(List<TeamDO> teamDOList) {
        if (CollUtil.isEmpty(teamDOList)) {
            log.warn("批量同步队伍到 ES 失败：队伍列表为空");
            return;
        }

        try {
            BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
            
            for (TeamDO teamDO : teamDOList) {
                if (teamDO == null || teamDO.getId() == null) {
                    continue;
                }
                
                TeamDocument teamDocument = convertToTeamDocument(teamDO);
                IndexOperation<TeamDocument> indexOp = IndexOperation.of(i -> i
                        .index(TEAM_INDEX)
                        .id(String.valueOf(teamDO.getId()))
                        .document(teamDocument)
                );
                bulkBuilder.operations(BulkOperation.of(o -> o.index(indexOp)));
            }

            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkBuilder.build());
            
            if (bulkResponse.errors()) {
                bulkResponse.items().forEach(item -> {
                    if (item.error() != null) {
                        log.error("批量同步队伍到 ES 失败：文档 ID={}，错误原因={}", 
                                item.id(), item.error().reason());
                    }
                });
            } else {
                log.info("批量同步队伍到 ES 成功，数量={}", teamDOList.size());
            }
        } catch (IOException e) {
            log.error("批量同步队伍到 ES 失败", e);
        }
    }

    @Override
    @Async
    public void batchDeleteUsersFromEs(List<Long> userIdList) {
        if (CollUtil.isEmpty(userIdList)) {
            log.warn("批量删除 ES 用户失败：用户ID列表为空");
            return;
        }

        try {
            BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
            
            for (Long userId : userIdList) {
                if (userId == null || userId <= 0) {
                    continue;
                }
                
                DeleteOperation deleteOp = DeleteOperation.of(d -> d
                        .index(USER_INDEX)
                        .id(String.valueOf(userId))
                );
                bulkBuilder.operations(BulkOperation.of(o -> o.delete(deleteOp)));
            }

            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkBuilder.build());
            
            if (bulkResponse.errors()) {
                bulkResponse.items().forEach(item -> {
                    if (item.error() != null) {
                        log.error("批量删除 ES 用户失败：文档 ID={}，错误原因={}", 
                                item.id(), item.error().reason());
                    }
                });
            } else {
                log.info("批量删除 ES 用户成功，数量={}", userIdList.size());
            }
        } catch (IOException e) {
            log.error("批量删除 ES 用户失败", e);
        }
    }

    @Override
    @Async
    public void batchDeleteTeamsFromEs(List<Long> teamIdList) {
        if (CollUtil.isEmpty(teamIdList)) {
            log.warn("批量删除 ES 队伍失败：队伍ID列表为空");
            return;
        }

        try {
            BulkRequest.Builder bulkBuilder = new BulkRequest.Builder();
            
            for (Long teamId : teamIdList) {
                if (teamId == null || teamId <= 0) {
                    continue;
                }
                
                DeleteOperation deleteOp = DeleteOperation.of(d -> d
                        .index(TEAM_INDEX)
                        .id(String.valueOf(teamId))
                );
                bulkBuilder.operations(BulkOperation.of(o -> o.delete(deleteOp)));
            }

            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkBuilder.build());
            
            if (bulkResponse.errors()) {
                bulkResponse.items().forEach(item -> {
                    if (item.error() != null) {
                        log.error("批量删除 ES 队伍失败：文档 ID={}，错误原因={}", 
                                item.id(), item.error().reason());
                    }
                });
            } else {
                log.info("批量删除 ES 队伍成功，数量={}", teamIdList.size());
            }
        } catch (IOException e) {
            log.error("批量删除 ES 队伍失败", e);
        }
    }

    /**
     * 将 UserDO 转换为 UserDocument（使用 BeanUtils 优化）
     */
    private UserDocument convertToUserDocument(UserDO userDO) {
        UserDocument document = new UserDocument();
        // 使用 BeanUtils 复制相同字段
        BeanUtils.copyProperties(userDO, document);
        
        // 解析 tags JSON 字符串为 List<String>
        List<String> tags = parseTags(userDO.getTags());
        document.setTags(tags);

        return document;
    }

    /**
     * 将 TeamDO 转换为 TeamDocument（包含重组后的 tags，使用 BeanUtils 优化）
     */
    private TeamDocument convertToTeamDocument(TeamDO teamDO) {
        TeamDocument document = new TeamDocument();
        // 使用 BeanUtils 复制相同字段
        BeanUtils.copyProperties(teamDO, document);

        // 重组队伍的 tags（合并所有成员的 tags）
        List<String> teamTags = rebuildTeamTags(teamDO.getId());
        document.setTags(teamTags);

        return document;
    }

    /**
     * 解析 tags JSON 字符串为 List<String>
     */
    private List<String> parseTags(String tagsJson) {
        if (!StringUtils.hasText(tagsJson)) {
            return new ArrayList<>();
        }

        try {
            return gson.fromJson(tagsJson, LIST_STRING_TYPE);
        } catch (Exception e) {
            log.warn("解析 tags JSON 失败：{}", tagsJson, e);
            return new ArrayList<>();
        }
    }

    /**
     * 重组队伍的 tags（合并所有成员的 tags，去重）
     * 1. 查询 user_team 表获取队伍所有成员ID（只查询未删除的关系）
     * 2. 查询 user 表获取每个用户的 tags
     * 3. 合并所有 tags 并去重
     *
     * @param teamId 队伍ID
     * @return 重组后的 tags 列表
     */
    private List<String> rebuildTeamTags(Long teamId) {
        // 1. 查询 user_team 表获取队伍所有成员ID（只查询未删除的关系）
        Set<Long> userIdSet = userTeamService.selectTeamUserIdList(teamId);
        if (userIdSet == null || userIdSet.isEmpty()) {
            log.debug("队伍 {} 没有成员，返回空 tags", teamId);
            return new ArrayList<>();
        }

        log.debug("队伍 {} 共有 {} 个成员，开始查询 tags", teamId, userIdSet.size());

        // 2. 批量查询所有成员的 tags（查询 user 表）
        Set<String> allTags = new HashSet<>();
        for (Long userId : userIdSet) {
            UserDO userDO = userMapper.selectById(userId);
            if (userDO != null && userDO.getIsDelete() != null && userDO.getIsDelete() == 0) {
                // 只处理未删除的用户
                if (StringUtils.hasText(userDO.getTags())) {
                    List<String> userTags = parseTags(userDO.getTags());
                    if (userTags != null && !userTags.isEmpty()) {
                        allTags.addAll(userTags);
                        log.debug("用户 {} 的 tags: {}", userId, userTags);
                    }
                }
            }
        }

        // 3. 转换为 List 并排序（便于去重和展示）
        List<String> result = allTags.stream()
                .sorted()
                .collect(Collectors.toList());

        log.info("队伍 {} 重组后的 tags: {} (共 {} 个)", teamId, result, result.size());
        return result;
    }
}

